package dkbta.server.domain;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Set;
import java.util.Vector;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import dkbta.Config;
import dkbta.server.domain.analyzer.QueryAnalyser;
import dkbta.server.domain.client.ClientManager;
import dkbta.server.domain.cu.CUStub;
import dkbta.server.domain.cu.CUStubsTable;
import dkbta.server.gui.Observable;
import dkbta.server.gui.Observer;
import dkbta.server.gui.forms.MonitoringForm;
import dkbta.server.util.exceptions.CoordinatorFailedToInitException;
import dkbta.server.util.exceptions.FailedToAnalyseException;
import dkbta.server.util.exceptions.FailedToDispatchPlans;
import dkbta.server.util.query.PlanPair;
import dkbta.server.util.query.QueriesTable;
import dkbta.server.util.query.Query;
import dkbta.server.util.query.Results;
import dkbta.util.Pair;





/**
 * @author Arkady Mishiev
 *
 */

public class Coordinator implements Observable, Runnable{
	private LinkedBlockingQueue<Query> m_queryInbox;
	private LinkedBlockingQueue<Results> m_queryOutbox;
	private volatile boolean m_shouldRun = true;
	private static Logger s_logger;
	private QueryAnalyser m_queryAnalyser;
	private Observer m_queriesPanelObserver;
	private Observer m_cuPanelObserver;
	private Observer m_logsPanelObserver;
	private ClientManager m_clientManager;
	private QueriesTable m_queriesTable;
	private CURegistrator m_cuRegistrator;
	private int m_registryPort;
	private ServerSocket m_registryServerSocket;
	private CUStubsTable m_stubsTable;

	public Coordinator() throws CoordinatorFailedToInitException {
		m_queryAnalyser = new QueryAnalyser();
		m_queryInbox = new LinkedBlockingQueue<Query>();
		m_queryOutbox = new LinkedBlockingQueue<Results>();
		m_clientManager = new ClientManager(m_queryInbox, m_queryOutbox);
		m_queriesTable = new QueriesTable(m_queryOutbox); /*queryId,query*/
		m_registryPort = Config.getInt("UNITS_REGISTRY_PORT", 9000);
		m_cuRegistrator = new CURegistrator();
		m_stubsTable = new CUStubsTable();
		m_shouldRun = true;					
	}

	private void init(){
		m_cuRegistrator.start();
		m_clientManager.start();
	}

	public void run(){
		s_logger.info("Starts ...");
		m_logsPanelObserver.updateLogsPanel("Coordination server starting ...");
		this.init();	
		while(m_shouldRun){
			try {
				s_logger.info("Waiting for query"); 
				Query query = m_queryInbox.take();
				s_logger.info("Received query. id :" + query.getId());
                                query.setState(Query.PlanState.ANALYZING);
                                Thread.sleep(3000);
                                m_queriesPanelObserver.updatePlansPanel(query);
				synchronized (m_stubsTable) {								
					Vector<String> stubsId = m_stubsTable.getAvailableStubs();
					if(stubsId.size() == 0){				
						Results res = new Results(query.getId(), 
								Config.FAILURE_PREFIX + 
								"There are no available Units.",
								query.getClientId());
						m_queryOutbox.put(res);
						s_logger.warn("Query rejected. There are no available units");
					}
					else{
						m_queryAnalyser.analyse(stubsId,query);				
						m_queriesTable.addQuery(query);
						dispatchPlans(query);
					}
				}
			} 
			catch (InterruptedException e) {				
				s_logger.warn("Failed to take query!");
			} 
			catch (FailedToAnalyseException e) {
				s_logger.warn("Failed to analyse query");
			}
			catch (FailedToDispatchPlans e) {
				s_logger.warn("Failed to dispatch plans");
			} 		
		}
		s_logger.info("Was stopped. Exits ...");
	}

	//TODO to move it into Query
	private void dispatchPlans(Query query) throws FailedToDispatchPlans{
		try{
			s_logger.info("Starts plans dispatching ...");
                        query.setState(Query.PlanState.DISPATCHING);
                        Thread.sleep(3000);
                        m_queriesPanelObserver.updatePlansPanel(query);
			Hashtable<String, PlanPair> allPlans = query.getPlans();
			String stubId;
			/*<plan,result>*/
			PlanPair plan;
			CUStub cuStub;
			Set<String> set = allPlans.keySet();
			Iterator<String> itr = set.iterator();
			while (itr.hasNext()) {
				stubId = itr.next();
				plan = allPlans.get(stubId);
				cuStub = m_stubsTable.get(stubId);
				/*<queryId,plan>*/
				Pair<String,String> stubsPlan = 
					new Pair<String, String>(query.getId(),plan.getPlan());
				cuStub.addPlan(stubsPlan,query);
			}
			s_logger.info("Dispatched ok");
		}
		catch (Exception e) {
			s_logger.warn("Failed to dispatch plan");
			e.printStackTrace();
			throw new FailedToDispatchPlans();			
		}
	}

	@Override
	public void addQueriesPanelObserver(Observer o){
		m_queriesPanelObserver = o;
                m_clientManager.addQueriesPanelObserver(o);
                m_stubsTable.addQueriesPanelObserver(o);
	}

	@Override
	public void addCUPanelObserver(Observer o) {
		m_cuPanelObserver = o;
		m_stubsTable.addCUPanelObserver(o);
	}

	@Override
	public void addLogsPanelObserver(Observer o) {
		m_logsPanelObserver = o;
		m_stubsTable.addLogsPanelObserver(o);
		m_clientManager.addLogsPanelObserver(o);
	}

	public static void main(String[] args) {
		try{
			s_logger = Logger.getLogger(Coordinator.class);
			PropertyConfigurator.configure("./files/conf/server.log4j.properties");					
			Coordinator coordinator = new Coordinator();		
			MonitoringForm gui = new MonitoringForm(coordinator);
		}
		catch(Exception e){
			System.err.println("The server failed on start");
			System.exit(-1);
		}
	}

	private class CURegistrator extends Thread {

		public CURegistrator() {
			super("CURegistrator");
		}

		public void run(){		
			while(m_shouldRun){
				if(m_registryServerSocket == null){
					try {
						m_registryServerSocket = new ServerSocket(m_registryPort);
						s_logger.info("Ready on port :" + m_registryPort);
						m_logsPanelObserver.updateLogsPanel("[CURegistrator] Ready on port :" 
								+ m_registryPort);
					} catch (IOException e) {
						s_logger.fatal("Could not listen on port: " + m_registryPort + 
						"\nClosing ...");
						m_logsPanelObserver.updateLogsPanel("UnitRegistry Could not listen on port: " + 
								m_registryPort + "\nClosing ...");
						try {
							Thread.sleep(5000);
						} catch (InterruptedException e1) {}
						System.exit(-1);
					}				
				}
				Socket cuStubSocket = null;
				try {
					cuStubSocket = m_registryServerSocket.accept();
					s_logger.info("CU accepted from : " + 
							cuStubSocket.getInetAddress().getHostAddress() + 
							":" + cuStubSocket.getPort());
					m_stubsTable.addStub(cuStubSocket,m_queriesTable);
				} catch (IOException e) {
					s_logger.warn("Accept failed.");
					//System.exit(1);
				}		     			
			}		
			try {
				m_registryServerSocket.close();
			} catch (IOException e) {
				s_logger.warn("Failed to close server socket");
			}
		}

	}

}
